package main

import (
	"fmt"
	"strings"
	"sync"
	"time"
)

// 发布订阅模型
// 发布／订阅（publish-and-subscribe）模型通常被简写为pub／sub模型。
// 在这个模型中，消息生产者成为发布者（publisher），
// 而消息消费者则称对应订阅者（subscriber），生产者和消费者是M：N的关系。
// 在传统生产者和消费者模型中，成果是将消息发送到一个队列中，
// 而发布/订阅模型则是将消息发布给一个主题。
// 为此，我们构建了一个名为pubsub的发布订阅模型支持包：

type (
	subscriber chan interface{}         // 订阅者管道
	topicFunc  func(v interface{}) bool // 主题为一个过滤器
)

type Publisher struct {
	m           sync.RWMutex             // 读写锁
	buffer      int                      // 订阅消息队列缓存
	timeout     time.Duration            // 发布超时时间
	subscribers map[subscriber]topicFunc // 订阅者信息
}

// 构建发布者，工厂模式
func NewPublisher(publishTimeout time.Duration, buffer int) *Publisher {
	return &Publisher{
		buffer:      buffer,
		timeout:     publishTimeout,
		subscribers: make(map[subscriber]topicFunc),
	}
}

// 添加能够订阅某个主题的订阅者
func (p *Publisher) SubscribeTopic(topic topicFunc) chan interface{} {
	ch := make(chan interface{}, p.buffer) //缓存管道
	p.m.Lock()
	p.subscribers[ch] = topic
	p.m.Unlock()
	return ch
}

// 退出订阅
func (p *Publisher) Evict(sub chan interface{}) {
	p.m.Lock()
	defer p.m.Unlock()
	delete(p.subscribers, sub)
	close(sub)
}

// 发布一个主题
func (p *Publisher) Publish(v interface{}) {
	// 支持并发发布
	p.m.RLock()
	defer p.m.RUnlock()

	var wg sync.WaitGroup
	for sub, topic := range p.subscribers {
		wg.Add(1)
		// 发布的方法
		go p.sendTopic(sub, topic, v, &wg)
	}
	wg.Wait()
}

// 关闭发布者对象，同时关闭所有订阅者管道
func (p *Publisher) Close() {
	// 支持并发关闭

	p.m.Lock()
	defer p.m.Unlock()

	for sub := range p.subscribers {
		delete(p.subscribers, sub)
		close(sub)
	}
}

// 发送主题，可以容忍一定的超时
func (p *Publisher) sendTopic(sub subscriber, topic topicFunc, v interface{}, wg *sync.WaitGroup) {
	defer wg.Done()
	// 判断是否有订阅者订阅内容
	if topic != nil && !topic(v) {
		return
	}
	select {
	case sub <- v:
	case <-time.After(p.timeout):
	}

}

// 添加能够订阅全部主题的订阅者
func (p *Publisher) Subercribe() chan interface{} {
	return p.SubscribeTopic(nil)
}
func main() {
	p := NewPublisher(100*time.Millisecond, 10) // 共享发布者，里面公用map 会有并发的问题
	defer p.Close()
	all := p.Subercribe()
	golang := p.SubscribeTopic(func(v interface{}) bool {
		if s, ok := v.(string); ok {
			return strings.Contains(s, "golang")
		}
		return false
	})

	p.Publish("hello, world!") // 发布 如果修改成 go p.Publish("hello, world!") 还要下面就会有并发问题
	p.Publish("hello, golang") // 发布 所以要用到并发安全问题《公用资源问题》发布可能就会有并发发布

	go func() {
		for msg := range all {
			fmt.Println("all:", msg)
		}
	}()

	go func() {
		for msg := range golang {
			fmt.Println("golang:", msg)
		}
	}()
	// 运行一定时间后退出
	time.Sleep(3 * time.Second)
}

// ====================================
// 关于上面的业务逻辑思路
// m每个订阅者维护一个属于自己的channel管道,同时对自己的管道，添加上对应的topic主题，
// 当发布者发布的主题内容的时候，会判断该内容的主题是否与订阅者主题相同的，如果有的话则返回true
// 然后把存在订阅的主题管道里面写入发布的内容，写入的时候有写入的超时时间，避免其他原因导致吸入超时
// 返回false的不做任何存储，
// 上面难点，topicFunc 动态的函数，简单说外部调用时候可以自己动态的配置，或者动态的写逻辑，该方法就是做动态发布内容匹配用的
// 发布结构体设计难点，1.考虑锁,2.考虑超时时间,3.考虑是否配置管道缓存,4.考虑发布者的相关信息使用的是k-v类型的map,
// map 的key 是channel的一种新式，而value 是一个func<所谓的过滤器>
// 关于对应的chan 和func 都外部都有定义 map可以来规定定义类型
// 关于map 容器操作要用到锁，避免出现资源争抢问题
